package com.atguigu.test;

import org.apache.flume.*;
import org.apache.flume.conf.Configurable;
import org.apache.flume.sink.AbstractSink;

import java.io.FileOutputStream;
import java.io.IOException;

/**
 * 将channel的数据发到哪里
 *
 * 将数据打印到控制台或者储存到文件，通过配置文件选择
 */
public class MySink extends AbstractSink implements Configurable {

    private boolean isFile;
    private String fileName;

    /**
     * 从Channel中拉取数据并处理
     * @return
     * @throws EventDeliveryException
     */

    public Status process() throws EventDeliveryException {
        Status status = null;

        //Sink要自己处理事务
        Channel channel = getChannel();
        Transaction transaction = channel.getTransaction();
        try {
            transaction.begin();
            //拉取事件并处理
            Event take = channel.take();
            storeSomeEvent(take);

            transaction.commit();
            status = Status.READY;
        } catch (Exception e) {

            transaction.rollback();
            status =Status.BACKOFF;
        } finally {
            transaction.close();
        }
        return status;
    }

    /**
     * 自定义如何处理事件
     * @param event
     */
    private void storeSomeEvent(Event event) throws IOException {

        if (isFile) {
            //开一个文件储存
            if (fileName == null) {
                throw new IOException("Output file not exists!");
            }
            FileOutputStream outputStream = new FileOutputStream(fileName, true);
            outputStream.write(event.getBody());
            outputStream.close();
        } else {
            System.out.write(event.getBody());
        }

    }

    /**
     * 通过配置文件定义Sink
     * @param context 配置文件
     */
    public void configure(Context context) {
        isFile = context.getBoolean("storeAsFile", false);
        fileName = context.getString("filename");
    }
}
